package cn.tecnova.utils

import cn.tecnova.bean.BaseFlowBean
import com.alibaba.fastjson.{JSON, JSONObject}
import com.google.gson.Gson
import org.apache.commons.lang.StringUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD

/**
  * description:过滤基础事务流封装类
  **/
object BaseFlowtopicUtils {

  def getBaseFlowBeanRDD(rdd:RDD[ConsumerRecord[String,String]], topicName:String) ={

    val baseFlowBeanRDD: RDD[BaseFlowBean] = rdd.filter(record => {
      val jsonObj: JSONObject = JSON.parseObject(record.value())
      val tName = jsonObj.getString("topic_name")
      topicName.equals(tName)
    }).map(record => {
      val gson = new Gson()
      val jsonObj: JSONObject = JSON.parseObject(record.value())
      val baseFlowBean: BaseFlowBean = gson.fromJson(record.value(), classOf[BaseFlowBean])
      if (StringUtils.isNotEmpty(jsonObj.getString("article_html"))) {
        val html = jsonObj.getString("article_html").replaceAll("\\s+", "")
        baseFlowBean.article_html = html
      }
      if (StringUtils.isNotEmpty(jsonObj.getString("article_content"))) {
        val content = jsonObj.getString("article_content").replaceAll("\\s+", "")
        baseFlowBean.article_content = content
      }
      baseFlowBean
    })

    baseFlowBeanRDD

  }
}
